package personal.wang.handler;

import com.google.protobuf.InvalidProtocolBufferException;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import personal.wang.abs.AbstractMessageHandler;
import personal.wang.domain.CommandMsg;
import personal.wang.inter.ExecuteMethod;
import personal.wang.manager.WorkerManager;
import personal.wang.message.Header;
import personal.wang.message.HeartbeatMessage;
import personal.wang.message.Message;
import personal.wang.enm.MessageType;
import personal.wang.service.CommandService;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * 采用注释 + 反射的方式来定义，各种命令由那个方法来处理。
 * 命令保存在 Message#content 中，使用 new String(content) 转化。
 * 命令的格式是
 * commad --args1 value --arg2 value2
 */
public class CommandMessageHandler extends AbstractMessageHandler {
    private static Logger logger = LoggerFactory.getLogger(CommandMessageHandler.class);
    private WorkerManager workerManager ;
    private CommandService commandService ;
    private Map<String, Method> keyMethod = new HashMap<>();
    @Override
    public void handler(Message message , ChannelHandlerContext ctx  ) {
        if(message.getHeader().getFlag() == MessageType.COMMAND_MESSAGE.getFlag()){
            // 广播模板
            // todo: 这里可以另外一种方式，如果一个模板中设置了 producer 的并发数量是 2 ，
            // 则可以让两个 worker 个执行一个 producer 的任务，
            // 这样方便控制总的测试数据的数量。
            // todo: 更深的一个功能点，是根据负载的情况，比较科学的调度。
            byte[] content = message.getContent();
            CommandMsg.CommandMessage commandMessage = null ;
            try {
                commandMessage = CommandMsg.CommandMessage.parseFrom(content);
            } catch (InvalidProtocolBufferException e) {
                throw new RuntimeException(e);
            }
            int messageCnt = commandMessage.getMessageCnt();
            String bootstrapServer = commandMessage.getBootstrapServer();
            logger.info("messageCnt:{} , bootstrapServer: {}" , messageCnt , bootstrapServer);
            workerManager.getWorkers().forEach((k,v)->{
                message.getHeader().setFlag(MessageType.TO_WORKER_COMMAND_MESSAGE.getFlag());
                v.getChannel().writeAndFlush(message);
            });
        }else{
            try {
                String cmd = new String(message.getContent(), StandardCharsets.UTF_8);
                logger.info("{}" , cmd);
                Object invoke = keyMethod.get(cmd).invoke(commandService);
                byte[] bytes = invoke.toString().getBytes(StandardCharsets.UTF_8);
                Header header = new Header(1 , 36
                        , bytes.length
                        , UUID.randomUUID().toString()
                        , MessageType.HEART_BEAT_REQUEST.getFlag());
                Message m = new HeartbeatMessage(header , bytes);
                ctx.writeAndFlush(m);
            } catch (IllegalAccessException e) {
                logger.error("{}" , e.getMessage());
            } catch (InvocationTargetException e) {
                logger.error("{}" , e.getMessage());
            }
        }
    }
    public CommandMessageHandler(WorkerManager workerManager){
        this.workerManager = workerManager;
        commandService = new CommandService();
        commandService.setWorkerManager(workerManager);
        for (Method method : commandService.getClass().getDeclaredMethods()) {
            if(method.isAnnotationPresent(ExecuteMethod.class)){
                ExecuteMethod annotation = method.getAnnotation(ExecuteMethod.class);
                method.setAccessible(true);
                keyMethod.put(annotation.tag() , method);
            }
        }
    }
    public void setWorkerManager(WorkerManager workerManager) {
        this.workerManager = workerManager;
    }

    public WorkerManager getWorkerManager() {
        return workerManager;
    }
}
